package com.study.kafka.producer;

import com.study.kafka.message.Demo01Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import java.util.concurrent.ExecutionException;

/**
 * @author yuantongqin
 * 2021/8/3
 **/
@Component
@Slf4j
public class Demo01Producer {

    @Autowired
    private KafkaTemplate<Object,Object> kafkaTemplate;

    public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException {
        // 创建一个message
        Demo01Message message = new Demo01Message();
        message.setId(id);
        ListenableFuture<SendResult<Object, Object>> send = kafkaTemplate.send(Demo01Message.TOPIC, message);
        send.addCallback(new SuccessCallback<SendResult<Object, Object>>() {
            @Override
            public void onSuccess(SendResult<Object, Object> result) {
                log.info("发送成功："+result.getProducerRecord().value());

            }
        }, new FailureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("发送失败：",ex);
                // 进行重试
            }
        });
        return send.get();
    }

    public ListenableFuture<SendResult<Object, Object>> asyncSend(Integer id){
        Demo01Message message = new Demo01Message();
        message.setId(id);
        ListenableFuture<SendResult<Object, Object>> send = kafkaTemplate.send(Demo01Message.TOPIC, message);
        return send;
    }
}
